package com.zhang.spark_1.spark_core.req

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * @title:
 * @author: zhang
 * @date: 2021/12/8 23:05 
 */
object Spark06_req3_pageflow {

  def main(args: Array[String]): Unit = {
    //获取spark的连接
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("top10")
    val sc: SparkContext = new SparkContext(conf)
    //读取数据
    val rdd: RDD[String] = sc.textFile("data/user_visit_action.txt")

    val actionDataRDD: RDD[UserVisitAction] = rdd.map(
      line => {
        val datas: Array[String] = line.split("_")
        UserVisitAction(
          datas(0),
          datas(1).toLong,
          datas(2),
          datas(3).toLong,
          datas(4),
          datas(5),
          datas(6).toLong,
          datas(7).toLong,
          datas(8),
          datas(9),
          datas(10),
          datas(11),
          datas(12).toLong
        )
      }
    )
    //计算分母
    val pageToCountMap: Map[Long, Long] = actionDataRDD.map(
      action => {
        (action.page_id, 1L)
      }
    ).reduceByKey(_ + _).collect().toMap

    //  计算分子
    //根据sessionID进行分组
    val sessionRDD: RDD[(String, Iterable[UserVisitAction])] = actionDataRDD.groupBy(_.session_id)

    //分组后根据时间进行排序（生序）
    val mvRDD: RDD[(String, List[((Long, Long), Int)])] = sessionRDD.mapValues(
      iter => {
        val sortLIst: List[UserVisitAction] = iter.toList.sortBy(_.action_time)
        //[1,2,3,4]=>[1-2,2-3,3-4]
        val flowIDs: List[Long] = sortLIst.map(_.page_id)
        val tuples: List[(Long, Long)] = flowIDs.zip(flowIDs.tail)
        tuples.map(
          tuples => {
            (tuples, 1)
          }
        )
      }
    )
    //结构转化（（pageID1，pageid2），1）
    val flowRDD: RDD[((Long, Long), Int)] = mvRDD.map(_._2).flatMap(list => list)
    //分组求和（（pageID1，pageid2），sum）
    val flowCount: RDD[((Long, Long), Int)] = flowRDD.reduceByKey(_ + _)

    //分子/分母 求页面的单挑转化率
    flowCount.foreach{
      case ((pageId1,pageId2), i) => {
        val l: Long = pageToCountMap.getOrElse(pageId1, 0l)
        println(s"页面$pageId1 跳转到页面 $pageId2 到条转化率为："+(i.toDouble/l))
      }
    }

    sc.stop()
  }

  //用户访问动作表
  case class UserVisitAction(
       date: String, //用户点击行为的日期
       user_id: Long, //用户的ID
       session_id: String, //Session的ID
       page_id: Long, //某个页面的ID
       action_time: String, //动作的时间点
       search_keyword: String, //用户搜索的关键词
       click_category_id: Long, //某一个商品品类的ID
       click_product_id: Long, //某一个商品的ID
       order_category_ids: String, //一次订单中所有品类的ID集合
       order_product_ids: String, //一次订单中所有商品的ID集合
       pay_category_ids: String, //一次支付中所有品类的ID集合
       pay_product_ids: String, //一次支付中所有商品的ID集合
       city_id: Long //城市 id
    )
}
